package cn.sciento.transfer.infra.rocketmq;

import cn.sciento.core.exception.CommonException;import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import cn.sciento.transfer.infra.rocketmq.config.MqConfigProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class RocketmqConsumersGroupManager {
  private final Logger logger = LoggerFactory.getLogger(getClass());
  
  private final Map<String, DefaultMQPushConsumer> consumerGroups = new HashMap<>();
  
  private final Map<String, Boolean> started = new HashMap<>();
  
  @Autowired
  public MqConfigProperties mqConfigProperties;
  
  public synchronized DefaultMQPushConsumer getOrCreateConsumer(String group, String topic, String tag) {
    if (this.consumerGroups.containsKey(group)) {
      return this.consumerGroups.get(group);
    }
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group);
    consumer.setNamesrvAddr(this.mqConfigProperties.getNameSrv());
    consumer.setInstanceName(Long.toString(System.currentTimeMillis()));
    consumer.setConsumeThreadMin(1);
    consumer.setConsumeThreadMax(1);
    try {
      consumer.subscribe(topic, tag);
    } catch (MQClientException e) {
      this.logger.error(e.getMessage(), e);
    } 
    this.consumerGroups.put(group, consumer);
    this.started.put(group, Boolean.FALSE);
    this.logger.info("RocketMQ CREATED for SCS group {} created", group);
    return consumer;
  }
  
  public synchronized void startConsumer(String group, MessageListenerConcurrently messageListenerOrderly) {
    start(group, messageListenerOrderly);
  }
  
  public synchronized void stopConsumer(String group) {
    stop(group);
  }
  
  public synchronized void restartConsumer(String group, String topic, String tag, MessageListenerConcurrently messageListenerConcurrently) {
    getOrCreateConsumer(group, topic, tag);
    start(group, messageListenerConcurrently);
  }
  
  private void stop(String group) {
    if (this.consumerGroups.get(group) != null) {
      this.consumerGroups.get(group).shutdown();
      this.consumerGroups.remove(group);
      this.started.remove(group);
    } 
  }
  
  private synchronized void start(String group, MessageListenerConcurrently messageListenerOrderly) {
    if (this.started.get(group) != null && this.started.get(group)) {
      return;
    }
    try {
      this.consumerGroups.get(group).registerMessageListener(messageListenerOrderly);
      this.consumerGroups.get(group).start();
      this.started.put(group, Boolean.TRUE);
    } catch (Exception e) {
      this.logger.error("RocketMQ Consumer hasn't been started. Caused by " + e
          .getMessage(), e);
      throw new CommonException(e);
    } 
  }
  
  public synchronized void suspend(String consumerGroup) {
    if (this.consumerGroups.get(consumerGroup) == null) {
      return;
    }
    if (!this.started.get(consumerGroup)) {
      return;
    }
    this.consumerGroups.get(consumerGroup).suspend();
  }
  
  public synchronized void resume(String consumerGroup) {
    if (this.started.get(consumerGroup) == null || !this.started.get(consumerGroup)) {
      return;
    }
    this.consumerGroups.get(consumerGroup).resume();
  }
  
  public synchronized boolean hasBeenStart(String group) {
    return (this.started.get(group) != null && this.started.get(group));
  }
  
  public synchronized Set<String> getConsumerGroups() {
    return this.consumerGroups.keySet();
  }
}
